"""
Copyright (c) 2022 Huawei Technologies Co.,Ltd.

openGauss is licensed under Mulan PSL v2.
You can use this software according to the terms and conditions of the Mulan PSL v2.
You may obtain a copy of Mulan PSL v2 at:

          http://license.coscl.org.cn/MulanPSL2

THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
See the Mulan PSL v2 for more details.
"""
"""
Case Type   : 套接字文件
Case Name   : 套接字文件无权限，连接数据库
Create At   : 2020/06/29
Owner       : n@ningyali
Description :
    1.获取套接字文件路径
    2.获取套接字文件名
    3.修改套接字文件权限为不可读不可写
    4.连接数据库
Expect      :
    1.成功
    2.成功
    3.成功
    4.连接失败
History     :
    Modified by @wan005 at 2022/01/20:还原listen_addresses增加localhost信息
    Modified by @li-xin12345 at 2023/03/15:研发代码变更，listen_addresses不允许置空，删除对应代码
"""

import os
import time
import unittest
from yat.test import Node
from yat.test import macro

from testcase.utils.Logger import Logger
from testcase.utils.Constant import Constant
from testcase.utils.Common import Common
from testcase.utils.CommonSH import *


class SocketNoPermission(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.log = Logger()
        cls.log.info(f'-----{os.path.basename(__file__)} start-----')
        cls.dbUserNode = Node(node='dbuser')
        cls.data_path = macro.DB_INSTANCE_PATH
        cls.com = Common()
        cls.Constant = Constant()
        cls.sh = CommonSH()

    def setUp(self):
        self.log.info('----------------------------检查数据库状态-----------------------------')
        db_status = self.sh.getDbClusterStatus('status')
        if not db_status:
            self.sh.startDbCluster()

            db_status = self.sh.getDbClusterStatus('status')
            if not db_status:
                raise Exception("db status is not true, please check!")
        self.log.info(db_status)

    # 方式一
    def test_1_socket_no_permission(self):
        global socket_file_path
        self.log.info('----------------------------test1:套接字文件无权限，连接数据库-----------------------------')
        self.log.info('----------------------------方式一：不修改配置参数，使用-h指定套接字文件路径连接-----------------------------')
        # 获取套接字文件路径
        get_socket_dir_cmd = f"show unix_socket_directory;"
        self.log.info(get_socket_dir_cmd)
        msg = self.sh.executDbSql(get_socket_dir_cmd)
        self.log.info(msg)
        msg = msg.splitlines()
        socket_dir = msg[-2].strip()
        self.log.info(socket_dir)

        # 获取套接字文件名
        get_file_list_cmd = f'ls -a {socket_dir}'
        self.log.info(get_file_list_cmd)
        file_list = self.dbUserNode.sh(get_file_list_cmd).result()
        self.log.info(file_list)
        socket_file_name = ''
        for file in file_list.split():
            self.log.info(file)
            if (self.dbUserNode.db_port not in file) or ('lock' in file):
                pass
            else:
                socket_file_name = file
        self.log.info(socket_file_name)

        # 修改套接字文件权限为不可读不可写
        socket_file_path = os.path.join(socket_dir, socket_file_name)
        chmod_cmd = f"chmod 100 {socket_file_path}"
        self.log.info(chmod_cmd)
        chmod_msg = self.dbUserNode.sh(chmod_cmd).result()
        self.log.info(chmod_msg)
        self.assertEqual(chmod_msg, '')

        # 通过socket连接数据库（预期失败）
        self.socket_conn_cmd = f'source {macro.DB_ENV_PATH};'
        self.socket_conn_cmd += f'gsql -d {self.dbUserNode.db_name} -p {self.dbUserNode.db_port} -h {socket_dir} -l'
        self.log.info(self.socket_conn_cmd)
        conn_msg = self.dbUserNode.sh(self.socket_conn_cmd).result()
        self.log.info(conn_msg)
        self.assertNotIn(self.Constant.CONN_DB_SUCCESS_MSG, conn_msg)

    # 方式二
    def test_2_socket_no_permission(self):
        self.log.info('----------------------------test2:套接字文件无权限，连接数据库-----------------------------')
        self.log.info('----------------------------方式二：不指定套接字文件路径连接-----------------------------')
        # 获取主机名
        cmd = 'cat /etc/hostname'
        self.log.info(cmd)
        hostname = self.dbUserNode.sh(cmd).result()
        self.log.info(hostname)

        # 修改套接字文件权限为不可读不可写
        chmod_cmd = f"chmod 100 {socket_file_path}"
        self.log.info(chmod_cmd)
        chmod_msg = self.dbUserNode.sh(chmod_cmd).result()
        self.log.info(chmod_msg)
        self.assertEqual(chmod_msg, '')

        # 不指定socket路径，通过socket连接数据库（预期失败）
        self.socket_conn_cmd = f'source {macro.DB_ENV_PATH};'
        self.socket_conn_cmd += f'gsql -d {self.dbUserNode.db_name} -p {self.dbUserNode.db_port} -l'
        self.log.info(self.socket_conn_cmd)
        conn_msg = self.dbUserNode.sh(self.socket_conn_cmd).result()
        self.log.info(conn_msg)
        self.assertNotIn(self.Constant.CONN_DB_SUCCESS_MSG, conn_msg)

    def tearDown(self):
        self.log.info(f'-----{os.path.basename(__file__)} end-----')
        # 还原套接字文件权限
        chmod_cmd = f"chmod 700 {socket_file_path}"
        self.log.info(chmod_cmd)
        chmod_msg = self.dbUserNode.sh(chmod_cmd).result()
        self.log.info(chmod_msg)
        self.assertEqual(chmod_msg, '')
